package com.tour.dal

import com.shujia.common.{Constants, SparkTool}
import org.apache.spark.sql.SparkSession

object CityWideApp extends SparkTool {
  /**
    *
    * 在子类中实现run方法，编写自定义的代码逻辑
    *
    * @param spark ： spark 入口
    */
  override def run(spark: SparkSession): Unit = {

    //将多次使用的表加载到内存中
    spark.sql(s"cache table ${Constants.DIM_DATABASE_NAME}.${Constants.DIM_ADMINCODE_TABLE_NAME}")

    spark.sql(
      s"""
         |
         |insert OVERWRITE table ${Constants.DAL_TOUR_DATABASE_NAME}.${Constants.CITY_WIDE_TABLE_NAME} partition(day_id=$day_id)
         |SELECT
         |  /*+ BROADCAST(a),BROADCAST(c),BROADCAST(d) */
         |	a.mdn,
         |	c.city_name AS d_city_name,
         |	d.city_name AS o_city_name,
         |	d.prov_name AS o_province_name,
         |	b.number_attr AS number_attr,
         |  case
         |    when a.d_max_distance >=10 and a.d_max_distance<50 then "[10-50)"
         |    when a.d_max_distance >=50 and a.d_max_distance<80 then "[50-80)"
         |    when a.d_max_distance >=80 and a.d_max_distance<120 then "[80-120)"
         |    when a.d_max_distance >=120 and a.d_max_distance<200 then "[120-200)"
         |    when a.d_max_distance >=200 and a.d_max_distance<400 then "[200-400)"
         |    when a.d_max_distance >=400 and a.d_max_distance<800 then "[400-800)"
         |    else "800~" end as d_max_distance,
         | case
         |    when a.d_stay_time>=3 and a.d_stay_time<6 then "[3-6)"
         |    when a.d_stay_time>=6 and a.d_stay_time<12 then "[6-12)"
         |    when a.d_stay_time>=12 and a.d_stay_time<24 then "[12-24)"
         |    when a.d_stay_time>=24 and a.d_stay_time<48 then "[24-48)"
         |    when a.d_stay_time>=48 and a.d_stay_time<72 then "[48-72)"
         |    else "72~" end as d_stay_time,
         |	b.gender AS gender,
         |	b.trmnl_brand AS trmnl_brand,
         |	b.packg AS pckg_price,
         |	b.conpot AS conpot,
         |	b.age AS age
         |FROM
         |	(
         |		SELECT
         |			*
         |		FROM
         |			${Constants.DAL_TOUR_DATABASE_NAME}.${Constants.CITY_TOURIST_TABLE_NAME}
         |		WHERE
         |			day_id = $day_id
         |	) AS a
         |JOIN (
         |	SELECT
         |		*
         |	FROM
         |		${Constants.DIM_DATABASE_NAME}.${Constants.DIM_USERTAG_TABLE_NAME}
         |	WHERE
         |		month_id = $month_id
         |) AS b ON a.mdn = b.mdn
         |JOIN (
         |	SELECT DISTINCT
         |		city_id,
         |		city_name
         |	FROM
         |		${Constants.DIM_DATABASE_NAME}.${Constants.DIM_ADMINCODE_TABLE_NAME}
         |) AS c ON a.d_city_id = c.city_id
         |JOIN ${Constants.DIM_DATABASE_NAME}.${Constants.DIM_ADMINCODE_TABLE_NAME} AS d ON a.source_county_id = d.county_id
         |
        |
      """.stripMargin)
  }
}
